package com.leilei.source.customize;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/9 20:39
 * @desc flink程序从Mysql中加载数据源
 */
public class FlinkSourceCustomizeByMysql {
    public static void main(String[] args) throws Exception {
        //准备环境 env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);
        //获取数据源 source
        DataStreamSource<VehicleAlarm> streamSource = env.addSource(new MysqlSource());
        //数据处理 todo
        streamSource.print();
        //数据收集 sink
        //程序执行 execute
        env.execute("mysql-source");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class VehicleAlarm {
        private String id;
        private String licensePlate;
        private String plateColor;
        private Long deviceTime;
        private String zone;
    }

    public static class MysqlSource extends RichParallelSourceFunction<VehicleAlarm> {
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet result = null;
        private boolean flag = true;
        String url = "jdbc:mysql://xxx:3306/alarm-sc?useUnicode=true&characterEncoding=utf-8&useSSL=false";

        @Override
        public void open(Configuration parameters) throws Exception {
            conn = DriverManager
                    .getConnection(url, "root", "xx..");
            String sql = "select id,license_plate,plate_color,device_time,`zone` from vehicle_alarm_202103";
            ps = conn.prepareStatement(sql);
            super.open(parameters);
        }


        @Override
        public void run(SourceContext<VehicleAlarm> ctx) throws Exception {
            while (flag) {
                result = ps.executeQuery();
                while (result.next()) {
                    String id = result.getString("id");
                    String licensePlate = result.getString("license_plate");
                    String plateColor = result.getString("plate_color");
                    Long deviceTime = result.getLong("device_time");
                    String zone = result.getString("zone");
                    VehicleAlarm vehicleAlarm = new VehicleAlarm(id, licensePlate, plateColor, deviceTime, zone);
                    ctx.collect(vehicleAlarm);
                }
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }


        @Override
        public void close() throws Exception {
            if (conn != null) {
                conn.close();
            }
            if (ps != null) {
                ps.close();
            }
            if (result != null) {
                result.close();
            }
        }
    }
}
